测试基础架构与 TaaS 设计

"""
TaaS 平台核心 API 设计（FastAPI 实现）
"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid
 
app = FastAPI(title="TaaS Platform API", version="1.0")
 
# ===== 数据模型 =====
class TestExecutionRequest(BaseModel):
    project_id: str
    suite_id: str
    environment: str  # "auto" 或指定环境 ID
    parallel: int = 4
    tags: Optional[list[str]] = None
    data_snapshot: Optional[str] = None
    notify: Optional[str] = None  # Slack/Email 频道
 
class TestExecutionResponse(BaseModel):
    execution_id: str
    status: str
    environment_id: str
    estimated_duration: int  # 秒
 
# ===== 核心接口 =====
@app.post("/api/v1/executions", response_model=TestExecutionResponse)
async def create_execution(req: TestExecutionRequest):
    """创建测试执行任务"""
    execution_id = str(uuid.uuid4())
 
    # 1. 获取或创建测试环境
    if req.environment == "auto":
        env = await provision_environment(req.project_id)
    else:
        env = await get_environment(req.environment)
 
    # 2. 准备测试数据
    if req.data_snapshot:
        await restore_snapshot(env, req.data_snapshot)
 
    # 3. 提交执行任务
    await submit_execution(
        execution_id=execution_id,
        suite_id=req.suite_id,
        env_id=env.id,
        parallel=req.parallel,
        tags=req.tags
    )
 
    return TestExecutionResponse(
        execution_id=execution_id,
        status="QUEUED",
        environment_id=env.id,
        estimated_duration=await estimate_duration(req.suite_id, req.parallel)
    )
 
@app.get("/api/v1/executions/{execution_id}")
async def get_execution(execution_id: str):
    """查询执行状态"""
    execution = await get_execution_status(execution_id)
    if not execution:
        raise HTTPException(404, "执行任务不存在")
    return execution
 
@app.get("/api/v1/executions/{execution_id}/report")
async def get_report(execution_id: str, format: str = "html"):
    """获取测试报告"""
    report_url = await generate_report(execution_id, format)
    return {"report_url": report_url}
 
@app.post("/api/v1/environments")
async def create_environment(project_id: str, template: str):
    """创建测试环境"""
    env = await provision_environment(project_id, template)
    return {"environment_id": env.id, "endpoint": env.endpoint}
 
@app.delete("/api/v1/environments/{env_id}")
async def destroy_environment(env_id: str):
    """销毁测试环境"""
    await teardown_environment(env_id)
    return {"status": "DESTROYED"}
 
@app.get("/api/v1/metrics")
async def get_metrics(project_id: str, days: int = 7):
    """获取测试度量数据"""
    return await query_metrics(project_id, days)

# 测试资产仓库结构
test-assets/
├── projects/
│   └── order-service/
│       ├── suites/                    # 测试套件
│       │   ├── smoke.yaml             # 冒烟测试套件
│       │   ├── regression.yaml        # 回归测试套件
│       │   └── e2e.yaml               # E2E 测试套件
│       ├── environments/              # 环境模板
│       │   ├── minimal.yaml           # 最小依赖环境
│       │   └── full-stack.yaml        # 全栈环境
│       ├── data/                      # 测试数据
│       │   ├── baseline/              # 基线数据快照
│       │   │   └── v1.0.0.sql
│       │   └── factories/             # 数据工厂脚本
│       │       └── order_factory.py
│       └── config/                    # 测试配置
│           ├── timeout.yaml
│           └── retry-policy.yaml